package com.doit.demo.day01;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * @DATE 2022/2/26/20:36
 * @Author MDK
 * @Version 2021.2.2
 *
 * 使用MapFunction和和RichMapFunction
 *
 *    RichMapFunction中的方法执行顺序（即Task的生命周期方法）
 *
 *    open(初始化时执行一次) -> map（每来一条数据执行一次） -> close(停止job时会执行一次)
 *
 *
 **/
public class MapDemo {
    public static void main(String[] args) throws Exception {

        //本地环境执行的默认并行度: 当前机器CPU的逻辑核数(CPU线程数)
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",8081);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        System.out.println("当前执行环境的并行度:"+env.getParallelism());
        System.out.println("----------------------------------------");

        DataStreamSource<String> lines = env.socketTextStream("linux01", 8888);
        System.out.println("socketTextStream生成的DataStream并行度为:"+lines.getParallelism());
        System.out.println("--------------------------------------------");

        //map是多并行的算子,如果没有指定该算子的并行度,默认与当前环境的并行度一致
        SingleOutputStreamOperator<String> streamOperator = lines.map(new MyMapFunction());
        /*System.out.println("streamOperator算子所在并行度为:"+streamOperator.getParallelism());*/

        //调用sink
        DataStreamSink<String> streamSink = streamOperator.addSink(new RichSinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
                int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
                System.out.println(indexOfThisSubtask + ">" + value);
            }
        });

        int parallelism = streamSink.getTransformation().getParallelism();
        System.out.println("sink算子所在的subtask的并行度:"+parallelism);

        env.execute();

    }

    private static class MyMapFunction extends RichMapFunction<String,String>{

        @Override
        public void open(Configuration parameters) throws Exception {
            //初始化连接或恢复之前的数据状态
            System.out.println("!!! open method invoked in subtask:"+getRuntimeContext().getIndexOfThisSubtask());
        }

        @Override
        public String map(String value) throws Exception {
            //执行程序
            System.out.println("### map method invoked in subtask:"+getRuntimeContext().getIndexOfThisSubtask());
            return value.toUpperCase();
        }

        @Override
        public void close() throws Exception {
            //关闭连接
            System.out.println("$$$ close method invoked in subtask:" + getRuntimeContext().getIndexOfThisSubtask());
        }


    }

}
